package com.cheng.im.service.massage.mq;

import com.cheng.im.common.bean.msg.ProtoMsgV1;
import com.cheng.im.common.constants.MQConstants;
import com.cheng.im.service.massage.service.impl.MessageSyncService;
import com.cheng.im.service.massage.service.impl.P2PMessageService;
import com.cheng.im.service.protobuilder.P2PMessageBuilder;
import com.google.protobuf.InvalidProtocolBufferException;
import io.swagger.v3.oas.annotations.parameters.RequestBody;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Slf4j
@Component
@RocketMQMessageListener(
        topic = MQConstants.IM_2_MSG_SERVICE,
        consumerGroup = "MSG_CONSUMER_GROUP"
)
public class ChatOperateReceiver implements RocketMQListener<Message> {

    @Resource
    P2PMessageService p2PMessageService;
    @Resource
    MessageSyncService messageSyncService;

    @Override
    public void onMessage(Message message) {
        ProtoMsgV1.Message pkg;
        try {
            pkg = ProtoMsgV1.Message.parseFrom(message.getBody());
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }

        ProtoMsgV1.HeadType headType = pkg.getType();

        // 下面判断各种消息类型，业务处理、消息接收ack、消息已读ack、消息撤回等等
        if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_REQUEST)) {
            p2PMessageService.process(pkg);
        } else if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_RECEIVE_ACK)) {
            // 接收端ack
            messageSyncService.receiveMark(pkg);
        } else if (headType.equals(ProtoMsgV1.HeadType.MESSAGE_READ)) {
            // 用户已读消息
            messageSyncService.readMark(pkg);
        }
    }
}
